package com.springboot.kafka.consumer;

import com.springboot.kafka.producer.RecordBean;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class ConsumerDemoInterceptor implements org.apache.kafka.clients.consumer.ConsumerInterceptor<String, RecordBean> {
    @Override
    public ConsumerRecords<String, RecordBean> onConsume(ConsumerRecords<String, RecordBean> records) {
        //poll方法返回去执行，可以对消息进行过滤
        Map<TopicPartition, List<ConsumerRecord<String, RecordBean>>> map = new HashMap<>();

        for (TopicPartition tp : records.partitions()) {
            List<ConsumerRecord<String, RecordBean>> recordList = records.records(tp);
            List<ConsumerRecord<String, RecordBean>> newRecordList =
                    recordList.stream()
                            //过滤规则
                            .filter(e -> (System.currentTimeMillis() - e.timestamp()) < 10 * 1000)
                            .collect(Collectors.toList());
            if (newRecordList != null) {
                map.put(tp, newRecordList);
            }
        }
        return new ConsumerRecords<>(map);
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        //提交消费位移后执行
        offsets.forEach((tp,offset)->{
            System.out.println("");
        });
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
